package com.xiao.dynamic_rule.datagen;

import com.alibaba.fastjson.JSON;
import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.utils.KafkaUtils;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;

public class ActionLogGen {
    public static void main(String[] args) {

        //创建多线程，并行造数
        for (int i=0;i<=1;i++){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    KafkaProducer<String, String> kafkaProducer = KafkaUtils.getKafakProducer("hadoop101:9092,hadoop102:9092,hadoop103:9092");

                    while (true){
                        LogBean logBean = new LogBean();

                        //生成账号 如： 009392
                        String account = StringUtils.leftPad(RandomUtils.nextInt(1, 10000) + "", 6, "0");
                        logBean.setAccount(account);
                        logBean.setAppId("com.xiao.xxxxx");
                        logBean.setAppVersion("1.15.1");
                        logBean.setCarrier("中国电信");
                        //deviceId直接用account
                        logBean.setDeviceId(account);
                        logBean.setIp("192.168.10.101");
                        logBean.setLatitude(RandomUtils.nextDouble(10.0,52.0));
                        logBean.setLongitude(RandomUtils.nextDouble(120.0,160.0));
                        logBean.setDeviceType("iphone12");
                        logBean.setNetType("5G");
                        logBean.setOsName("ios");
                        logBean.setOsVersion("15.5");
                        logBean.setResolution("2048*1024");
                        logBean.setReleaseChannel("浏览器");
                        logBean.setEventId(RandomStringUtils.randomAlphabetic(1).toUpperCase());

                        HashMap<String, String> properties = new HashMap<>();
                        for (int i = 0;i < RandomUtils.nextInt(1,5);i++){
                            //生成属性 如： p1=v3,p3=v4.......
                            properties.put("p"+RandomUtils.nextInt(1,10),"v" + RandomUtils.nextInt(1,10));
                        }

                        logBean.setProperties(properties);
                        logBean.setTimeStamp(System.currentTimeMillis());
                        logBean.setSessionId(RandomStringUtils.randomNumeric(10,10));

                        //将日志对象转成json
                        String log = JSON.toJSONString(logBean);
                        //System.out.println(log);

                        //不能太快，线程sleep一下
                        try {
                            Thread.sleep(RandomUtils.nextInt(500,501));
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        //写入kafka的topic中
                        ProducerRecord<String, String> applog = new ProducerRecord<>("applog", log);
                        kafkaProducer.send(applog);
                    }
                }
            }).start();
        }

    }
}
